package fregat.connector.channel;

import fregat.base.util.FileParser;
import fregat.base.util.Logger;
import fregat.base.net.Protocol;
import fregat.connector.ConnectorService;
import fregat.connector.base.FTPTransfer;
import fregat.connector.base.FileDescription;
import fregat.connector.base.FileHandler;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.util.Enumeration;

/**
 * AbstractChannel that retrieves data from an input file(or stdin) and writes data
 * to an output file(or stdout).
 *
 * @author Alexander Mann <br><b><i>ECO_L O G</i> &nbsp Logistiksysteme GmbH</b>
 * @version 1.0
 */
public class FileChannel extends AbstractChannel {

  private FileHandler input; // input file handler
  private FileHandler output; // output file handler
  private PrintStream out; // output stream
  private BufferedReader in; // input reader
  private boolean shutdown; // flag whether shutdown was initiated
  private FileParser fileParser; // reads a file and parses each line
  private FTPTransfer ftp; // transfers data and controls file from or to a FTP server
  private String beginDelimiter; // begin of file delimiter
  private String endDelimiter; // end of file delimiter
  private Protocol protocol;

  /**
   * Creates a new file channel.
   *
   * @param name Configuration name
   */
  public FileChannel(String name) {
    super(name);

    // configuration -->
    String configPath = "channel.config." + name;
    beginDelimiter = ConnectorService.getConfiguration().getConfigurationString(configPath + ".delimiter.begin");
    endDelimiter = ConnectorService.getConfiguration().getConfigurationString(configPath + ".delimiter.end");
    ftp = new FTPTransfer(configPath);
    input = new FileHandler(name, FileHandler.IN, ftp);
    output = new FileHandler(name, FileHandler.OUT, ftp);
    setProtocol(ConnectorService.getConfiguration().configureProtocol(name));
    // <--
  }

  public void setProtocol(Protocol protocol) {
    this.protocol = protocol;
  }

  /**
   * Writes data to an output file.
   *
   * @param data
   * @return True, if write was success
   */
  public boolean send(byte[] data) {
    if (output.isReady()) {
      if (out == null) {
        if (output.isStandardIO()) {
          // writing to standard out
          out = System.out;
        } else {
          // create new data file
          out = output.createDataFile();
        }
      }

      if (out != null) {
        // write data
        String str = new String(data);
        output.log(str);
        if (protocol != null) {
          return protocol.send(out, str.getBytes());
        } else {
          out.println(str);
          return true;
        }
      }
    }
    return false;
  }

  /**
   * Called by worker, when all data was processed.
   */
  public void dataComplete() {
    if (out != null) {
      // data was written and control file is configured

      synchronized (output) {
        // disconnect from worker
        connected(false);
        output.setReady(false);

        // close data file
        out.close();
        out = null;

        // write control file
        if(output.isControlled()) {
          output.writeControlFile();
        }

        // store file(s) on FTP server
        if (ftp.isConfigured()) {
          output.store();
        }

        // move a copy of the local data and control file to another directory if configured
        // and delete local data and control file
        output.moveFiles();
      }
    }

    if(in == null && !output.isControlled()) {
      shutdown();
    }
  }

  /**
   * Starts the connected listeners and reading of input file.
   */
  public void start() {
    super.start();

    if (output.isConfigured()) {
      if (!output.isStandardIO()) {
        if (output.isControlled()) {
          // a control file is used for the output file --> periodic checks
          new FilePoller(output, false) {
            protected void process() {
              output.reset();
              // delete (local) output file if still existing
              output.deleteDataFile(false);
              output.setReady(true);
              connected(true);
            }
          }.start();
        } else {
          // writing a single run to an output file
          output.reset();
          output.deleteDataFile(false);
          output.setReady(true);
        }
      } else {
        // writing a single run to standard out
        connected(true);
      }
    }

    if (input.isConfigured()) {
      if (input.isStandardIO()) {
        // reading from standard in
        new Thread() {
          public void run() {
            if (in == null) {
              in = new BufferedReader(new InputStreamReader(System.in));
            }

            // wait some time for other processes to start up as system.out blocks
            try {
              Thread.sleep(1000);
            } catch (InterruptedException ex) {
            }

            String input = "";
            try {
              do {
                input = in.readLine();
                if (input != null && !input.equals("") && !shutdown) {
                  received(input.getBytes());
                }
              } while (input != null && !input.equals("") && !shutdown);
              in.close();
            } catch (IOException ex) {
              ConnectorService.error(getName() + ", FileChannel.start", ex);
            }
          }
        }.start();
      } else if (input.isConfigured()) {
        // reading from file
        new FilePoller(input, true) {
          protected void process() {
            // create timestamp for backup file
            fileHandler.log(Logger.getTimestamp());

            // get file(s) from FTP server
            if (ftp.isConfigured()) {
              fileHandler.retrieve();
            }

            // read from file(s)
            Enumeration e = fileHandler.dataFiles();
            while (e.hasMoreElements() && !shutdown) {
              FileDescription file = (FileDescription) e.nextElement();
              if (file.wasTransferred()) {
                fileParser = new FileParser() {
                  protected boolean parseLine(String line) {
                    try {
                      fileHandler.log(line);
                      received(line.getBytes());
                      return true;
                    } catch (Exception ex) {
                      ConnectorService.error(getName() + ", FileChannel.start", ex);
                    }
                    return false;
                  }
                };

                markLimit(file, false);
                fileParser.readFile(file.toString());
                markLimit(file, true);
              }
            }

            // delete data file
            fileHandler.deleteDataFile(true);

            // delete control file
            fileHandler.deleteControlFile(true);
            fileHandler.setReady(false);
          }
        }.start();

        if (!input.isControlled() && !output.isControlled()) {
          shutdown();
        }
      }
    }
  }

  private void markLimit(FileDescription file, boolean eof) {
    String delimiter = null;
    if (beginDelimiter != null && !eof) {
      delimiter = beginDelimiter;
    } else if (endDelimiter != null && eof) {
      delimiter = endDelimiter;
    }

    if (delimiter != null) {
      received(delimiter.replaceFirst("%file", file.toString()).getBytes());
    }
  }

  /**
   * Stops all connected listeners and then closes files.
   */
  public void shutdown() {
    // close file processing
    if (fileParser != null) {
      fileParser.shutdown();
    }
    shutdown = true;

    // close input handler
    if (input != null) {
      input.shutdown();
    }

    // close output handler
    if (output != null) {
      output.shutdown();
    }

    super.shutdown();
  }

  public String toString() {
    StringBuffer buffer = new StringBuffer();
    if (input.isConfigured()) {
      buffer.append(super.toString());
      buffer.append(" ");
      buffer.append(ConnectorService.resource.getString("FileChannel_toString_1").replaceFirst("%1", input.logFiles()));
      buffer.append(" ");
      buffer.append(input.toString());
    }
    if (output.isConfigured()) {
      if (input.isConfigured()) {
        buffer.append("\r\n");
      }
      buffer.append(super.toString());
      buffer.append(" ");
      buffer.append(ConnectorService.resource.getString("FileChannel_toString_2").replaceFirst("%1", output.logFiles()));
      buffer.append(" ");
      buffer.append(output.toString());
    }
    return buffer.toString();
  }

  /**
   * Checks if a file is (not) existing and calls the method @link process.
   */
  private abstract class FilePoller extends Thread {
    FileHandler fileHandler;
    boolean readyWhenExisting;

    public FilePoller(FileHandler file, boolean readyWhenExisting) {
      this.fileHandler = file;
      this.readyWhenExisting = readyWhenExisting;
    }

    public void run() {
      do {
        if (fileHandler.isControlled()) {
          boolean state;

          synchronized (fileHandler) {
            // check for control file
            state = fileHandler.isReady();

            // move a copy of the local data and control file to another directory if configured
            // and delete local data and control file
            fileHandler.moveFiles();

            boolean controlFileExists = fileHandler.isControlFileExisting();

            boolean updatedState = !(controlFileExists ^ readyWhenExisting);

            fileHandler.setReady(updatedState);

            // RMA20060404: bei ausgehenden Files zunächst process aufrufen
            if (fileHandler.isReady() && state != fileHandler.isReady() && !readyWhenExisting) {
              // state has changed to ready -> process data
              process();
            }            
          }

          // RMA20060404: process in Abhaengigkeit von readyWhenExisting vor oder nach Channelstatus-Behandlung
          // aufrufen
          // bei Inputfiles (readyWhenExisting = true) alles wie gehabt zunächst channelstatus umsetzen und
          //                                           dann file bearbeiten
          // bei Outputfiles (readyWhenExisting=false) darf ChannelStatus erst nach process gemacht werden
          //                                           denn in process wird ja auch ein existierendes Datenfile
          //                                           gelöscht

          //  RMA20060404: bei eingehenden Files erst hier process aufrufen
          if (fileHandler.isReady() && state != fileHandler.isReady() && readyWhenExisting) {
            // state has changed to ready -> process data
            process();
          }

          fileHandler.timeout();

        } else {
          // no control file -> process data and exit
          process();
        }
      } while (!shutdown && fileHandler.isControlled());
    }

    protected abstract void process();
  }
}
